package a3;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

public class HbaseJava {
    public static void main(String[] args) throws Exception {

        Connection conn;
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        conf.set("hbase.zookeeper.quorum", "jikehadoop01,jikehadoop02");
        conn= ConnectionFactory.createConnection(conf);
        Admin admin = conn.getAdmin();

        //define namespace:tablename
        TableName tableName = TableName.valueOf("guaguagua:student");
        String tableNameString = tableName.toString();

        //create namespace if not existed
        if(tableNameString.contains(":")){
            String nameSpace = tableNameString.substring(0,tableNameString.indexOf(":"));
            try {
                NamespaceDescriptor namespaceDescriptor = admin.getNamespaceDescriptor(nameSpace);
                System.out.println("namespaceDescriptor : " + namespaceDescriptor.getName() + " existed. ");
            } catch (NamespaceNotFoundException e){
                admin.createNamespace(NamespaceDescriptor.create(nameSpace).build());
                System.out.println("we created new namespace: " + nameSpace);
            }
        }

        if(admin.tableExists(tableName)) {
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
            System.out.println(tableName.toString() + " existed, we are deleting it now.");
        } else {
            HTableDescriptor htd = new HTableDescriptor(tableName);
            htd.addFamily(new HColumnDescriptor(Bytes.toBytes("info")));
            htd.addFamily(new HColumnDescriptor(Bytes.toBytes("score")));
            admin.createTable(htd);
            System.out.println("table created");
        }

        // disable table
        // admin.disableTable(TableName.valueOf("tablename"));
        // truncate table
        // admin.truncateTable(TableName.valueOf("tablename"), true);
        // delete table
        // admin.deleteTable(TableName.valueOf("tablename"));

        //insert data into table
        try (Table table = conn.getTable(TableName.valueOf("guaguagua:student"))) {
            StudentInfo si1 = new StudentInfo("Tom","20210000001","01","95","100");
            StudentInfo si2 = new StudentInfo("guaguagua","G20210735010450","05","95","100");

            // insert data into table
            insertIntoTable(si1, "row1",table);
            insertIntoTable(si2, "row2",table);

            System.out.println("all data inserted.");
        }

    }

    private static void insertIntoTable(StudentInfo si, String rowNum, Table table) throws Exception {
        Put put = new Put(Bytes.toBytes(rowNum));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes(si.getName()));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("student_id"), Bytes.toBytes(si.getStudentId()));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("class"), Bytes.toBytes(si.getStudentClass()));
        put.addColumn(Bytes.toBytes("score"), Bytes.toBytes("understanding"), Bytes.toBytes(si.getUnderstanding()));
        put.addColumn(Bytes.toBytes("score"), Bytes.toBytes("programming"), Bytes.toBytes(si.getProgramming()));
        table.put(put);
        table.close();
        System.out.println("row data of " + si.getName() + " inserted");
    }

}
